package com.ebupt.migu.ksms.controller;

import java.security.MessageDigest;
import java.time.Instant;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import com.ebupt.migu.common.pojo.Topic;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ebupt.migu.common.sms.CallerHungUpMessageBean;
import com.ebupt.migu.common.sms.MegamarketRequestBean;
import com.ebupt.migu.common.util.DateUtil;
import com.ebupt.migu.common.util.StrategyConstants;
import com.ebupt.migu.ksms.utils.HTTPUtil;
import com.google.common.collect.Maps;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@Service
public class CallerHungUpMessageController {

    @Value("${megamarket.ip}")
    private String MEGAMARKET_IP;
    @Value("${megamarket.saltValue}")
    private String MEGAMARKET_SALTVALUE;
    @Value("${megamarket.accessCode}")
    private String MEGAMARKET_ACCESSCODE;
    @Value("${megamarket.sendMarketingSMS}")
    private String MEGAMARKET_SENDMARKETINGSMS;
    @Autowired
    private HTTPUtil httpUtil;
    @Autowired
    private RedisTemplate redisTemplate;

    @KafkaListener(topics = "RFMENT-CALLER-HUNG-UP-MESSAGE-TOPIC", groupId = "RFMENT-CALLER-HUNG-UP-MESSAGE-GROUP")
    @Retryable(value = Exception.class, maxAttempts = 3, backoff = @Backoff(delay = 1000))
    public void callerHungUpMessage(JSONObject object, Acknowledgment ack) {
        CallerHungUpMessageBean caller = JSON.parseObject(object.toJSONString(), CallerHungUpMessageBean.class);
        if (caller.getSendEndTime() < Instant.now().getEpochSecond()) {
            log.info("超出最晚下发时间，请求作废：{}", caller.getMegamarket());
            ack.acknowledge();
            return;
        }
        JSONObject sendParams = new JSONObject();
        sendParams.put("body", getBody(caller.getMegamarket().getPort(), caller.getMegamarket().getMsisdn(),
            caller.getMegamarket().getTemplateId(), caller.getMegamarket().getContent()));
        sendParams.put("header", getHeader(caller.getMegamarket().getStrategyId()));
        try {
            int i =
                caller.getMegamarket().getMsisdn() == null ? 0 : caller.getMegamarket().getMsisdn().split(",").length;
            httpUtil.doPostSingleThread(MEGAMARKET_IP + MEGAMARKET_SENDMARKETINGSMS, sendParams.toJSONString());
            // 发送数量统计
            String key = "caller-send-count-" + DateUtil.getDateToday() + "-" + caller.getMegamarket().getStrategyId();
            incrementRedisCounter(key, i);
            ack.acknowledge();
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }

    @KafkaListener(topics = Topic.OFFLINE_SMS_1130, groupId = "RFMENT-OFFLINE-SMS-GROUP-1130")
    public void offline1130(JSONObject object, Acknowledgment ack) {
        sendMsg(object, ack, Topic.OFFLINE_SMS_1130);
    }

    @KafkaListener(topics = Topic.OFFLINE_SMS_1131, groupId = "RFMENT-OFFLINE-SMS-GROUP-1131")
    public void offline1131(JSONObject object, Acknowledgment ack) {
        sendMsg(object, ack, Topic.OFFLINE_SMS_1131);
    }

    @KafkaListener(topics = Topic.OFFLINE_SMS_TEST, groupId = "RFMENT-OFFLINE-SMS-GROUP-TEST")
    public void offlineTest(JSONObject object, Acknowledgment ack) {
        sendMsg(object, ack, Topic.OFFLINE_SMS_TEST);
    }

    /**
     * 离线短信下发
     * @param ack
     */
    private void sendMsg(JSONObject object, Acknowledgment ack, String topic){
        try {
            MegamarketRequestBean bean = JSON.parseObject(object.toJSONString(), MegamarketRequestBean.class);
            // 添加日志追踪requestId
            MDC.put(Topic.REQUEST_ID_KEY,bean.getRequestId());
            log.info("下发topic --> {}",topic);
            long rep_ack_start = System.currentTimeMillis();
            ack.acknowledge();
            long rep_ack_end = System.currentTimeMillis();
            log.info("上报偏移量耗时：{}ms", rep_ack_end - rep_ack_start);
            JSONObject sendParams = new JSONObject();
            sendParams.put("body", getBody(bean.getPort(), bean.getMsisdn(), bean.getTemplateId(), bean.getContent()));
            sendParams.put("header", getHeader(bean.getStrategyId()));
            String sid = bean.getStrategyId();
            // 短信发送进度
            String key_incr = sid + "_" + DateUtil.getDateToday() + "_incr";
            // 成功数
            String key_success = sid + "_" + DateUtil.getDateToday() + "_success";
            long start = System.currentTimeMillis();
            httpUtil.doPostSingleThread(MEGAMARKET_IP + MEGAMARKET_SENDMARKETINGSMS, sendParams.toJSONString());
            long end = System.currentTimeMillis();
            log.info("本次请求耗时：{}ms", end - start);
            int i = Arrays.asList(bean.getMsisdn().split(",")).size();
            incrementRedisCounter(key_incr, i);
            incrementRedisCounter(key_success, i);
            long redis_incr_end = System.currentTimeMillis();
            log.info("Redis下发统计耗时：{}ms", redis_incr_end - end);
        } catch (Exception e) {
            log.error(e.getMessage());
            e.printStackTrace();
        }
        MDC.remove(Topic.REQUEST_ID_KEY);
    }

    private JSONObject getBody(int id, String msisdn, String templateId, String content) {
        HashMap<String, String> megamarket = getMegamarketConf(id);
        log.info("大营销平台端口配置信息：{}", megamarket);
        JSONObject body = new JSONObject();
        body.put("sendChannel", megamarket.get("sendChannel"));
        body.put("sendType", megamarket.get("sendType"));
        body.put("destPort", megamarket.get("destPort"));
        body.put("destSubPort", megamarket.get("destSubPort"));
        body.put("receiverNo", msisdn);
        if ("mms".equals(megamarket.get("sendType"))) {
            body.put("msgType", "1");
            body.put("msgContent", templateId);
        }
        if ("sms".equals(megamarket.get("sendType"))) {
            Map<String, String> params = new HashMap<>();
            params.put("templateId", templateId.equals("20070003") ? "20070003" : "20070001");
            body.put("params", params);
            body.put("msgType", StrategyConstants.MEGAMARKET_MSGTYPE);
            body.put("msgContent", content);
        }
        return body;
    }

    private Map<String, String> getHeader(String strategyId) {
        Map<String, String> header = Maps.newHashMap();
        header.put("version", StrategyConstants.MEGAMARKET_VERSION);
        header.put("transactionId", strategyId);
        header.put("timestamp", DateUtil.parseDateToStr(new Date(), DateUtil.DATE_TIME_FORMAT_YYYYMMDDHHMISS));
        header.put("accessCode", MEGAMARKET_ACCESSCODE);
        header.put("signature", getSignature(header));
        return header;
    }

    private String getSignature(Map<String, String> header) {
        StringBuilder signature = new StringBuilder();
        signature.append("version").append("=").append(header.get("version")).append("&");
        signature.append("transactionId").append("=").append(header.get("transactionId")).append("&");
        signature.append("timestamp").append("=").append(header.get("timestamp")).append("&");
        signature.append("accessCode").append("=").append(header.get("accessCode"));
        signature.append(MEGAMARKET_SALTVALUE);
        return getMD5String(signature.toString()).toLowerCase();
    }

    private static String getMD5String(String str) {
        try {
            MessageDigest md = MessageDigest.getInstance("MD5".toUpperCase());
            byte[] digest = md.digest(str.getBytes());
            return bytesToString(digest);
        } catch (Exception e) {
            log.error("md5加密失败，错误原因：[{}]", e.getMessage());
        }
        return null;
    }

    private static String bytesToString(byte[] data) {
        char[] hexDigits = {'0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f'};
        char[] temp = new char[data.length * 2];
        for (int i = 0; i < data.length; i++) {
            byte b = data[i];
            temp[(i * 2)] = hexDigits[(b >>> 4 & 0xF)];
            temp[(i * 2 + 1)] = hexDigits[(b & 0xF)];
        }
        return new String(temp).toUpperCase();
    }

    private HashMap<String, String> getMegamarketConf(int id) {
        return JSON.parseObject(
            String.valueOf(redisTemplate.opsForValue().get("mega-market-conf-" + String.valueOf(id))), HashMap.class);
    }

    private void incrementRedisCounter(String key, int i) {
        redisTemplate.opsForValue().increment(key, i);
    }
}
